package com.hlz.flink.chapter06.window;

import com.hlz.flink.chapter06.Stock;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * 事件时间窗口
 *
 * @author Hongliang Zhu
 * @create 2023-06-23 16:18
 */
public class WindowAggregationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> dataStreamSource = env.socketTextStream("127.0.0.1", 9696);
        SingleOutputStreamOperator<Stock> stockStream = dataStreamSource.filter(StringUtils::isNotEmpty)
                .map(new MapFunction<String, Stock>() {
                    @Override
                    public Stock map(String value) throws Exception {
                        final String[] split = value.split(",");
                        return new Stock(split[0], Double.parseDouble(split[1]), split[2], Long.parseLong(split[3]));
                    }
                });

        // 设置水位线
        SingleOutputStreamOperator<Stock> stockSingleOutputStreamOperator1 = stockStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Stock>forBoundedOutOfOrderness(Duration.ofMillis(2)).withTimestampAssigner(new SerializableTimestampAssigner<Stock>() {
                    @Override
                    public long extractTimestamp(Stock element, long recordTimestamp) {
                        return element.getTimeStamp();
                    }
                }));

//        SingleOutputStreamOperator<Stock> stockSingleOutputStreamOperator = stockSingleOutputStreamOperator1.keyBy(Stock::getId)
////                .window(TumblingEventTimeWindows.of(Time.milliseconds(10)))
////                .reduce(new ReduceFunction<Stock>() {
////                    @Override
////                    public Stock reduce(Stock value1, Stock value2) throws Exception {
////                        long max = Math.max(value1.getTimeStamp(), value1.getTimeStamp());
////                        return new Stock(value1.getId(), value1.getPrice() + value2.getPrice(), value1.getStockName(), max);
////                    }
////                });


        SingleOutputStreamOperator<String> aggregate = stockSingleOutputStreamOperator1.keyBy(Stock::getId)
                .window(TumblingEventTimeWindows.of(Time.milliseconds(10)))
                .aggregate(new AggregateFunction<Stock, Double, String>() {

                    /**
                     * 创建一个累加器，这就是为聚合创建了一个初始状态，每个聚
                     * 合任务只会调用一次。
                     * @return
                     */
                    @Override
                    public Double createAccumulator() {
                        System.out.println("初始化累加器");
                        return 0.0;
                    }

                    @Override
                    public Double add(Stock value, Double accumulator) {
                        System.out.println("add 方法调用...");
                        return value.getPrice() + accumulator;
                    }

                    @Override
                    public String getResult(Double accumulator) {
                        System.out.println("getResult 方法调用...");
                        return accumulator.toString();
                    }

                    /**
                     * 只在session窗口中生效
                     * @param a
                     * @param b
                     * @return
                     */
                    @Override
                    public Double merge(Double a, Double b) {
                        System.out.println("merge 方法调用...");

                        return null;
                    }


                });

        aggregate.print("window_agg");

        env.execute();
    }
}

